package com.bml.distributed.lock.distributedlock.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * zk分布式锁
 */
public class ZkLockImpl implements ZkLock, InitializingBean {
    private Logger log = LoggerFactory.getLogger(ZkLockImpl.class);

    //分布式锁的根节点,需要事先创建
    private final String LOCK_ROOT_PATH = "/ZkLock";

    private Map<String, CountDownLatch> concurrentMap = new ConcurrentHashMap<>();

    private final String CHAR_FIX = "/";

    @Autowired
    private CuratorFramework curatorFramework;

    private final Lock lock = new ReentrantLock();

    @Override
    public boolean lock(String lockpath) {

        //lockpath是调用获取锁传入的
        boolean result = false;
        String keyPath = LOCK_ROOT_PATH + lockpath;
        if (!lockpath.startsWith(CHAR_FIX)) {
            keyPath = LOCK_ROOT_PATH + CHAR_FIX + lockpath;
        }
        log.info("keypath="+keyPath);
        try {
            curatorFramework.create()
                    .creatingParentsIfNeeded()
                    //临时节点
                    .withMode(CreateMode.EPHEMERAL)
                    //不适用ACL
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(keyPath);
            //创建成功表示获取锁
            log.info("--------------获取锁成功,线程=[{}],lockpath=[{}]", lockpath, Thread.currentThread().getName());
            result = true;
        } catch (Exception e) {
            log.info("Thread:{};failed to acquire mutex lock for path:{}", Thread.currentThread().getName(), keyPath);
            e.printStackTrace();

            if (!concurrentMap.containsKey(lockpath)) {

                try {
                    lock.lock();
                    //不存在这个路径的map，然后加入到map，但是需要加锁，因为只有第一个县城能获取锁
                    /*
                     * 这里考虑到高并发场景,必须保证对同一个节点加锁的线程失败后是落在同一个countdown对象上
                     * ,否则有的线程永远没有办法唤醒了
                     */
                    //双重检查，进入获取zk锁失败的线程只有第二个能进入if，其他进来如果不 判断，那么就会覆盖countdown，那么有些线程就无法唤醒了
                    if (!concurrentMap.containsKey(lockpath)) {
                        concurrentMap.put(lockpath, new CountDownLatch(1));
                    }
                } catch (Exception exception) {
                    exception.printStackTrace();
                } finally {
                    lock.unlock();
                }

                CountDownLatch countDownLatch = concurrentMap.get(lockpath);
                if (countDownLatch != null) {
                    try {
                        //没有获取锁的阻塞，等待获取锁的zk删除节点出发监听
                        countDownLatch.await();
                    } catch (InterruptedException interruptedException) {
                        interruptedException.printStackTrace();
                        log.info("InterruptedException message:{}", interruptedException.getMessage());

                    }
                }
            }
        }
        return result;
    }

    /**
     * 释放锁
     *
     * @param lockpath
     * @return
     */
    @Override
    public boolean unlock(String lockpath) {
        String keyPath = LOCK_ROOT_PATH + lockpath;
        if (!lockpath.startsWith(CHAR_FIX)) {
            keyPath = LOCK_ROOT_PATH + CHAR_FIX + lockpath;
        }
        try {
            if (curatorFramework.checkExists().forPath(keyPath) != null) {
                curatorFramework.delete().forPath(keyPath);
            }
        } catch (Exception exception) {
            exception.printStackTrace();
            log.error("failed to release mutex lock");
            return false;
        }
        return true;
    }

    @Override
    public void afterPropertiesSet() throws Exception {

        //初始化调用
       // curatorFramework = curatorFramework.usingNamespace("zklock-namespace");
        //zk锁的根路径 不存在则创建
        try {
            if (curatorFramework.checkExists().forPath(LOCK_ROOT_PATH) == null) {
                //创建持久节点
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(LOCK_ROOT_PATH);
            }

            //添加节点的监听
            addWatcher(LOCK_ROOT_PATH);

        } catch (Exception exception) {
            log.error(exception.getMessage());
        }

    }
    /**
     * 监听节点事件
     * @param lockPath
     *       加锁的路径
     */
    private void addWatcher(String lockPath) throws Exception {
        String keyPath;
        if(LOCK_ROOT_PATH.equals(lockPath)){
            keyPath = lockPath;
        }else{
            keyPath = LOCK_ROOT_PATH + lockPath;
        }

        PathChildrenCache cache = new PathChildrenCache(curatorFramework,keyPath,false);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        /**
         * 添加监听器
         */
        cache.getListenable().addListener((client,event) -> {
            //子节点的删除事件
            if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                //获取删除路径的节点，比如/ZkLock/aaa
                String oldPath = event.getData().getPath();
                log.info("oldPath delete:{},缓存已经更新！",oldPath);
                //如果被删除的节点是包含了根节点，那么可能是锁释放了，那么要把计数器减下来，释放阻塞
                if(oldPath.contains(lockPath)){
                    //TODO 释放计数器，释放锁
                    CountDownLatch countDownLatch = concurrentMap.remove(oldPath);
                    if(countDownLatch != null){//有可能没有竞争,countdown不存在
                        countDownLatch.countDown();
                    }
                }
            }
        });
    }

}
